package com.asiainfo.dacp.datastash.broker.facade;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.asiainfo.dacp.datastash.broker.core.StashUnitModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitSchedType;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnit;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitDefine;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitSchedule;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;
import com.asiainfo.dacp.jdbc.JdbcTemplate;
import com.asiainfo.dacp.metamodel.domain.ds.model.MetaDataSource;
import com.asiainfo.dacp.util.StringUtils;
import com.asiainfo.dacp.util.vfs.VfsUtils;
import com.google.common.base.Strings;

import org.apache.commons.vfs2.FileObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cglib.beans.BeanMap;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

/**
 * Unit触发条件判断
 * <p>
 * 根据依赖类型完成依赖的判断
 * 1.文件标识，文件名，文件内容
 * 2.表标识，表名（数据源），SQL（数据源）
 * 3.时间依赖 ？ 这个需要重新设计一下，不使用这套
 *
 * @author MeiKefu
 * @date 2017-11-21
 */
@Service
public class UnitDependFacade {

    @Autowired
    private StashTraceLogger tracelog;
    private String clzName = this.getClass().getSimpleName();


    @Autowired
    private StashUnitModel stashUnitModel;

    public StashUnit fetchWaitingStashUnit(){
        String unitProcId = stashUnitModel.fetchWaitingUnitProcId();
        if(unitProcId!=null){
            return stashUnitModel.getStashUnit(unitProcId);
        }else{
            return null;
        }
    }

    public void requeue(StashUnit stashUnitStatus){
        stashUnitModel.requeue(stashUnitStatus);
    }

    /**
     * unit是否依赖到达
     */
    public boolean arrivalJuding(StashUnit stashUnit) {
        tracelog.debug(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDepServ:entry","进入依赖到达判断"));
        boolean isArrival = false;

        if(stashUnit==null){
            tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"fetchStashUnit:error","获取StashUnit失败"));
            return false;
        }

        StashUnitDefine stashUnitDefine = null;
        try {
            stashUnitDefine = stashUnit.getStashUnitDefine();
        } catch (Exception e) {
            tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"fetchStashUnit:error",e.getMessage()));
        }

        if(stashUnitDefine==null){
            return false;
        }

        StashUnitSchedule stashUnitSched = stashUnitDefine.getStashUnitSched();
        
        /**
         * fix: 如果任务没配置依赖，则跳过依赖判断.
         * */
        if(stashUnitSched == null){
        	return true;
        }

        if(stashUnitSched==null){
            tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"stateFile:null","依赖判断的内容为空"));
            return false;
        }

        if(UnitSchedType.File.value().equals(stashUnitSched.getSchedType())){
            FileObject fileObject = null;
            try {
                String fileName = stashUnitSched.getStateExpr();
                MetaDataSource dataSource = BrokerContext.getInstance().getDsDefs().get(stashUnitSched.getDsName());
                fileName = dataSource.getVfsHeader()+StringUtils.variableSubstitution(fileName,stashUnit);
                tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDepServ","判断文件到达：fileName"+fileName));
                fileObject = VfsUtils.getInstance().resolveFile(fileName);
                if (fileObject.exists()) {
                    tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDepServ:arrived","文件依赖到达判断成功"));
                    isArrival=true;
                }else{
                    tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDep:fail","文件依赖到达判断失败"));
                }
            } catch (Exception e) {
                tracelog.warn(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDep:error","文件依赖到达判断错误"+e.getMessage()));
            } finally {
                VfsUtils.getInstance().closeFileObject(fileObject);
            }

        }else if (UnitSchedType.SQL.value().equals(stashUnitSched.getSchedType())){
            MetaDataSource dataSource = BrokerContext.getInstance().getDsDefs().get(stashUnitSched.getDsName());
            String sql = stashUnitSched.getStateExpr();
            if(!Strings.isNullOrEmpty(sql)){
            	sql = StringUtils.variableSubstitution(sql, BeanMap.create(stashUnit).getBean());
            }
            tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDepServ","判断SQL语句：SQL"+sql+",dbName"+stashUnitSched.getDsName()+",dsDBName:"+dataSource.getDsDbName()));
            JdbcTemplate jt = new JdbcTemplate(dataSource.getDsDbName());
            List list =  jt.queryForList(sql);
            if(list.size()>0){
                tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDepServ:arrived","SQL依赖到达判断成功"));
                isArrival=true;
            }else{
                tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDep:fail","SQL依赖到达判断失败"));
            }
        }else{
            tracelog.error(StashTraceLogVo.buildStashTraceLogVo(stashUnit, clzName,"unitDepServ:error","不支持的调度类型"));
        }

        return isArrival;
    }
}
